//
// Created by ta&to on 2022/8/22.
//

#include "http_thread.h"

volatile long taskid = 0;

static inline void *thread_cycle(void *data);

// 获取当前线程数
static inline int thread_get_threads_num(thread_pool_t* tpt)
{
    rwlock_rdlock(&tpt->thread_rwlock);
    int num = tpt->threads;
    (void)  rwlock_unlock(&tpt->thread_rwlock);
    return num;

}
//获取空闲线程数
static inline int thread_get_freethreads_num(thread_pool_t* tpt)
{
    rwlock_rdlock(&tpt->thread_rwlock);
    int num = tpt->freethreads;
    (void)  rwlock_unlock(&tpt->thread_rwlock);
    return num;

}

static inline void thread_add_threads_num(thread_pool_t* tpt, int num){
    rwlock_wrlock(&tpt->thread_rwlock);
    atomicIncr(tpt->threads, num);
    (void) rwlock_unlock(&tpt->thread_rwlock);
}

static inline noder thread_rb_sel_node(Leader *leader, void *key)
{
    noder node = mapSel(leader, key);
    return node;
}
static inline int thread_rb_add_node(thread_pool_t *tpt, Leader *leader, void *key, void *value)
{
    add_linklist_after(&tpt->list_thread, key);
    RBtreenodeadd(leader, key, value);
    atomicIncr(tpt->threads, 1);
    atomicIncr(tpt->freethreads, 1);             // 新创建的线程空闲线程数加1
}
static inline int thread_rb_del_node(thread_pool_t *tpt, Leader *leader, void *key, table_t *t)
{
    destry_linklist(&tpt->list_thread, t);
    mapDel(leader, key);
    atomicDecr(tpt->threads, 1);
    atomicDecr(tpt->freethreads, 1);            // 线程被清除空闲线程数加1
}
static inline int thread_rb_mod_node(Leader *leader, void *key, void *value)
{
    treenodemodify(leader, key, value);
}
static inline int thread_rb_del_node_safe(thread_pool_t *tpt, Leader *leader, void *key, table_t *t)
{
    rwlock_wrlock(&tpt->thread_rwlock);
    thread_rb_del_node(tpt, leader, key, t);
    (void) rwlock_unlock(&tpt->thread_rwlock);
}
/*
 * 创建一个任务容器
 */
static inline thread_task_t *newtaskcontainer(void (*handler)(void *args, thread_task_args_t *args_t))
{
    thread_task_t *ttt;
    ttt = (thread_task_t *)http_calloc(sizeof(thread_task_t));
    ttt->next = NULL;
    ttt->handler = handler;
    ttt->taskid = taskid;
    atomicIncr(taskid, 1);
}
/**
 * 销毁任务容器
 */
static inline void destrytaskcontainer(thread_pool_t* tpt, thread_task_t *ttt)
{
    if(ttt) http_free(ttt);
}

/**
 * 加入任务
 * @param tpt
 * @param handler
 * @param args
 * @return
 */
static inline int Thread_Insert_Task_Container(thread_pool_t* tpt, void (*handler)(void *args, thread_task_args_t *args_t), void *args)
{
    thread_task_t *tt;
    if(NULL != (tt = newtaskcontainer(handler)))
    {
        tt->ctx = args;
        if(!task_push(&tpt->th_queue, tt))
            return thread_err;
        return thread_ok;
    }
    return thread_err;
}

/**
 * 向线程池队列插入任务
 * @param tpt
 * @param handler
 * @param args
 * @return
 */
static inline int Thread_Insert_Task_To_Pool(thread_pool_t* tpt, void (*handler)(void *args, thread_task_args_t *args_t), void *args)
{
    rwlock_wrlock(&tpt->task_rwlock);
    if(!Thread_Insert_Task_Container(tpt, handler, args))
    {
        (void) rwlock_unlock(&tpt->task_rwlock);
        return thread_err;
    }
    pthread_cond_signal(&tpt->th_sync_cond);
    (void) rwlock_unlock(&tpt->task_rwlock);
    return thread_ok;
}
// 任务入队
int Thread_Task_Post(thread_pool_t* tpt, void (*handler)(void *args, thread_task_args_t *args_t), void *args)
{
    int err;
    if(ISNULL(tpt)) return thread_err;
    if((err = Thread_Insert_Task_To_Pool(tpt, handler, args)) == thread_ok)
        Thread_Pool_Thread_Shortage(tpt);
    return err;
}

static inline thread_task_t* Thread_Get_Task_From_Pool(thread_pool_t* tpt)
{
    return task_pop(&tpt->th_queue);
}
// 任务出队
thread_task_t* Thread_Task_Pop(thread_pool_t* tpt)
{
    if(ISNULL(tpt)) return NULL;
    if(!rwlock_wrlock(&tpt->task_rwlock))        // 虽然是读取数据但是由于牵扯到指针移动所以使用写锁
        return NULL;
    thread_task_t *task = Thread_Get_Task_From_Pool(tpt);
    rwlock_unlock(&tpt->task_rwlock);
    return task;
}

//任务队列中是否有数据
static inline int hasTask(thread_pool_t *tpt)
{
    rwlock_rdlock(&tpt->task_rwlock);
    if(has_queue_next(&tpt->th_queue))
    {
        rwlock_unlock(&tpt->task_rwlock);
        return thread_ok;
    }
    rwlock_unlock(&tpt->task_rwlock);
    return thread_err;
}

static inline long Thread_Get_Task_Number(struct Queue *queue)
{
    return get_queue_size(queue);
}
// 获取等待执行任务数
static inline long Thread_Get_Task_Number_From_Pool(thread_pool_t *tpt)
{
    long num = 0;
    rwlock_rdlock(&tpt->task_rwlock);
    num = Thread_Get_Task_Number(&tpt->th_queue);
    (void) rwlock_unlock(&tpt->task_rwlock);
    return num;
}
/*
 * 销毁已经执行完毕的任务
 */
int Thread_Task_Destry(thread_pool_t* tpt, thread_task_t *ttt)
{
    if(ISNULL(tpt) || ISNULL(ttt)) return thread_err;
    if(!rwlock_wrlock(&tpt->task_rwlock))        // 虽然是读取数据但是由于牵扯到指针移动所以使用写锁
        return thread_err;
    destrytaskcontainer(tpt, ttt);
    rwlock_unlock(&tpt->task_rwlock);
    return thread_ok;
}

static int before_create_thread(pthread_attr_t *attr) {
    int err;
    if ((err = pthread_attr_init(attr) != 0))
    {
        fprintf(stderr, "pthread_attr_init failed.\n");
    }
    if ((err = pthread_attr_setschedpolicy(attr, SCHED_RR) != 0))
    {
        if(err == ENOTSUP)
        {
            fprintf(stdout, "pthread_attr_setschedpolicy SCHED_RR no suppert.\n");
        }else{
            fprintf(stderr, "pthread_attr_setschedpolicy SCHED_RR failed.\n");
        }
    }
    return thread_ok;
}

static int after_create_thread(pthread_attr_t *attr) {
    pthread_attr_destroy(attr);
    return thread_ok;
}

static int exchange_thread_status(thread_pool_t *tpt, noder node, enum run_state runstate) {
    thread_container_t * tct = (thread_container_t *)node->spot->value;
    tct = (thread_container_t *)node->spot->value;
    if(tct->runstate != runstate){
        tct->runstate = runstate;
        thread_rb_mod_node(&tpt->leader_thread, &tct->tid, tct);
    }
    return thread_ok;
}
/**
 * 回调任务中的方法
 * @param args
 * @param args_t
 */
static void interrupt_call(void *args, thread_task_args_t *args_t) {
    args_t->ttt->handler(args, args_t);
}

int Thread_Pool_Commit(thread_pool_t *tpt)
{
    int err;
    pthread_t tid;
    pthread_attr_t attr;
    before_create_thread(&attr);
    for (int i = 0; i < (tpt->initthreads > tpt->maxthreads?tpt->maxthreads:tpt->initthreads); ++i) {
        mutex_lock(&tpt->th_add_lock);
        err = pthread_create(&tid, &attr, thread_cycle, tpt);
        if (err)
            return thread_err;
    }
    after_create_thread(&attr);
}
static void cleanup_handler(thread_task_args_t *args) {        // 线程终止后的处理函数防止pthread_cond_wait占用锁导致程序死锁
    thread_pool_t *tpt = args->tpt;
    if(ISNOTNULL(args->tct)) {
        http_free(args->tct);
    }
    if(ISNOTNULL(args)){
        http_free(args);
    }
    mutex_unlock(&tpt->th_sync_mtx);
}

static inline void *thread_cycle(void *data){
    thread_task_args_t *args_t;
    thread_pool_t *tpt = (thread_pool_t*)data;
    pthread_t tid;
    noder node;
    thread_container_t *tct;
    pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, NULL);
    pthread_setcanceltype(PTHREAD_CANCEL_ASYNCHRONOUS, NULL);
    tid = pthread_self();
    char ttid[sizeof(char) * http_numlen(tid)+1];
    memset(ttid, 0, sizeof(ttid));
    http_sprintf(ttid, "%l", tid);
    rwlock_wrlock(&tpt->thread_rwlock);
    mutex_unlock(&tpt->th_add_lock);
    node = thread_rb_sel_node(&tpt->leader_thread, ttid);
    if(ISNULL(node)){
        args_t = (thread_task_args_t *) http_calloc(sizeof(thread_task_args_t));
        tct = (thread_container_t *) http_calloc(sizeof(thread_container_t));
        tct->tid = tid;
        tct->runstate = TP_THREAD_STATE_RUNNABLE;
        tct->freeresiduetimeout = http_times();              // 这里其实应该是在运行状态 但是为了满足新增线程都是空闲这一策略 依然赋予空闲状态， 很快该状态被修改为运行态
        thread_rb_add_node(tpt, &tpt->leader_thread, ttid, tct);
    } else {
        exchange_thread_status(tpt, node, TP_THREAD_STATE_RUNNABLE);
    }
    args_t->tpt = tpt;
    args_t->tct = tct;
    rwlock_unlock(&tpt->thread_rwlock);
    for (;;) {
        while (!hasTask(tpt)) {                  // 防止线程被意外唤醒
            rwlock_wrlock(&tpt->thread_rwlock);
            tct->freeresiduetimeout = http_times();          // 由于没有找到任务立刻将当前线程设为空闲状态并赋予时间戳
            node = thread_rb_sel_node(&tpt->leader_thread, ttid);
            if(ISNULL(node))
            {
                (void) rwlock_unlock(&tpt->thread_rwlock);
                return NULL;
            }
            exchange_thread_status(tpt, node, TP_THREAD_STATE_BLOCKED);
            (void) rwlock_unlock(&tpt->thread_rwlock);
            pthread_cleanup_push(cleanup_handler, args_t);
            mutex_lock(&tpt->th_sync_mtx);
wait:       if (pthread_cond_wait(&tpt->th_sync_cond, &tpt->th_sync_mtx)) //如果在调用线程尚未锁定互斥锁的情况下调用pthread_cond_wait()，则程序具有未定义的行为。 任何事情都可能发生，包括崩溃。
            {
                (void) mutex_unlock(&tpt->th_sync_mtx);
                return NULL;
            }
            (void) mutex_unlock(&tpt->th_sync_mtx);
            pthread_cleanup_pop(0);
            rwlock_wrlock(&tpt->thread_rwlock);
            if(tct->runstate == TP_THREAD_STATE_TERMINATED)
            {
                printf("thread is dead!");
                pthread_cond_signal(&tpt->th_sync_cond);
                goto wait;
            }
            node = thread_rb_sel_node(&tpt->leader_thread, ttid);           // 为防止node被改变这里在线程被唤醒后从新获取
            exchange_thread_status(tpt, node, TP_THREAD_STATE_RUNNABLE);
            tct->freeresiduetimeout = -1;               // 空闲时间戳为-1表示当前线程正在使用中
            (void) rwlock_unlock(&tpt->thread_rwlock);
        }
        thread_task_t *task = Thread_Task_Pop(tpt); // 从队列中取出一个任务
        if(ISNULL(task)) continue;
        args_t->ttt = task;
        args_t->args = task->ctx;
        atomicDecr(tpt->freethreads, 1);            // 空闲线程数减1
        interrupt_call(task->ctx, args_t);
        Thread_Task_Destry(tpt, task);                          // 销毁任务
        atomicIncr(tpt->freethreads, 1);             // 空闲线程数加1
    }
}

static inline table_t * select_one_thread_outtime(noder desn, thread_pool_t *tpt){
    if(ISNULL(desn) || ISNULL(tpt))return NULL;
    rwlock_wrlock(&tpt->thread_rwlock);
    ready_linklist_data(&tpt->list_thread);
    while(has_linklist_next(&tpt->list_thread))
    {
        table_t *t = get_linklist(&tpt->list_thread);
        noder node = mapSel(&tpt->leader_thread, t->value);
        long time = http_times();
        if(ISNOTNULL(node) && (((thread_container_t *)node->spot->value)->freeresiduetimeout != -1)
           &&(time - ((thread_container_t *)node->spot->value)->freeresiduetimeout) > tpt->threadresiduetimes
           && ((thread_container_t *)node->spot->value)->runstate == TP_THREAD_STATE_BLOCKED)
        {
            ((thread_container_t *)node->spot->value)->runstate = TP_THREAD_STATE_TERMINATED;
            treenodecopy(desn, node);
            (void) rwlock_unlock(&tpt->thread_rwlock);
            return t;
        }
    }
    (void) rwlock_unlock(&tpt->thread_rwlock);
}
/**
 * 任务检测线程自动扩充
 * 说明：该方法用于向任务队列插入任务时调用，如果任务数查过阈值，则自动创建线程
 */
void Thread_Pool_Thread_Shortage(thread_pool_t *tpt){
    int err;
    pthread_t tid;
    pthread_attr_t attr;
    if(Thread_Get_Task_Number_From_Pool(tpt) * THREAD_FACTOR > thread_get_freethreads_num(tpt))       // 当线程数量不够时
    {
        before_create_thread(&attr);
        mutex_lock(&tpt->th_add_lock);
        if(tpt->maxthreads > thread_get_threads_num(tpt)) {
            err = pthread_create(&tid, &attr, thread_cycle, tpt);
            if (err){
                mutex_unlock(&tpt->th_add_lock);
                return ;
            }
        }else mutex_unlock(&tpt->th_add_lock);
    }
}
/**
 * 空闲线程监控
 * @param tpt
 * @return
 */
void *Thread_Pool_Monitor_Daemon(thread_pool_t *tpt)
{
    int err;
    noder node;
    int res_kill;
    pthread_t tid;
    pthread_attr_t attr;
    thread_container_t *tct;
    noder dnode = (noder)http_malloc(sizeof(struct Node));          // 该对象可重复利用不必被回收
    before_create_thread(&attr);
    for(;;){
        http_time_sleep(1);
        if(Thread_Get_Task_Number_From_Pool(tpt) == 0 && tpt->minthreads < tpt->threads)  // 当没有任务并且允许的最小线程数小于当前线程数时
        {
            treenodenull(dnode);
            table_t *tb = select_one_thread_outtime(dnode, tpt);
            if (ISNULL(dnode)|| ISNULL(dnode->key) || ISNULL(dnode->spot)) continue;
            thread_container_t *ct = (thread_container_t *) dnode->spot->value;
            if (ISNULL(ct)) continue;
            res_kill = pthread_kill(ct->tid, 0);
            if (res_kill == ESRCH){
                fprintf(stderr, "the specified thread did not exists or already quit\n");
            }
            else if (res_kill == EINVAL)
                fprintf(stderr, "signal is invalid\n");
            else {
                int res = pthread_cancel(ct->tid);
                if (res != 0) {
                    fprintf(stderr, "stop thread-%d fail!\n", (int) ct->tid);
                } else {
                    thread_rb_del_node_safe(tpt, &tpt->leader_thread, dnode->key, tb);
                    fprintf(stdout, "stop thread-%d success!\n", (int) ct->tid);
                }
            }
        }
    }
}

pthread_t Thread_Pool_Monitor(thread_pool_t *tpt){
    pthread_t tid;
    pthread_create(&tid,NULL,(void *(*)(void *))Thread_Pool_Monitor_Daemon,(void *)tpt);
    return tid;
}

void Thread_Pool_Init(thread_pool_t* tpt)
{
    tpt->name = "ta&to";
    tpt->initthreads = THREAD_INIT_NUMBER;
    tpt->maxthreads = THREAD_MAX_NUMBER;
    tpt->minthreads = THREAD_MIN_NUMBER;
    tpt->threads = 0;
    tpt->freethreads = 0;
    tpt->threadresiduetimes = THREAD_RESIDUE_TIMES;
    pthread_mutex_init(&tpt->th_sync_mtx, NULL);
    pthread_mutex_init(&tpt->th_add_lock, NULL);
    pthread_rwlock_init(&tpt->thread_rwlock, NULL);
    pthread_rwlock_init(&tpt->task_rwlock, NULL);
    pthread_cond_init(&tpt->th_sync_cond, NULL);
    initLeader(&tpt->leader_thread);
    initLinkList(&tpt->list_thread);
    initQueue(&tpt->th_queue, THREAD_TASK_MAX_QUEUE);
}